package pub.cleangao.memm.registry;

import com.google.protobuf.ByteString;
import com.ibm.etcd.api.Event;
import com.ibm.etcd.api.KeyValue;
import com.ibm.etcd.api.RangeResponse;
import com.ibm.etcd.client.EtcdClient;
import com.ibm.etcd.client.KvStoreClient;
import com.ibm.etcd.client.kv.KvClient;
import com.ibm.etcd.client.kv.WatchUpdate;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.stream.Collectors;


@Slf4j
public class EtcdServiceRegistry extends AbstractRpcServiceRegistry {


    private static EtcdServiceRegistry instance;


    private EtcdServiceRegistry() {
    }

    public static EtcdServiceRegistry getInstance() {
        if (instance == null) {
            instance = new EtcdServiceRegistry();
        }
        return instance;
    }


    private static KvClient kvClient = null;

    static {
        // etcd 初始化
        KvStoreClient client = null;
        if (RpcRuntime.getConfigProperties().getRegistryUsername() != null
                && RpcRuntime.getConfigProperties().getRegistryPassword() != null
                && !RpcRuntime.getConfigProperties().getRegistryUsername().isBlank()
                && !RpcRuntime.getConfigProperties().getRegistryPassword().isBlank()) {
            client = EtcdClient
                    .forEndpoints(RpcRuntime.getConfigProperties().getRegistryUrl())
                    .withCredentials(RpcRuntime.getConfigProperties().getRegistryUsername(), RpcRuntime.getConfigProperties().getRegistryPassword())
                    .withPlainText()
                    .build();
        } else {
            client = EtcdClient
                    .forEndpoints(RpcRuntime.getConfigProperties().getRegistryUrl())
                    .withPlainText().build();
        }

        kvClient = client.getKvClient();
        KvClient.WatchIterator watchIterator = kvClient.watch(ByteString.copyFrom(ETCD_KEY_PREFIX, StandardCharsets.UTF_8)).asPrefix().start();
        async.execute(() -> {
            while (watchIterator.hasNext()) {
                WatchUpdate watchUpdate = watchIterator.next();
                if (watchUpdate != null) {
                    List<Event> events = watchUpdate.getEvents();
                    if (events != null && !events.isEmpty()) {
                        g:
                        for (Event e : events) {
                            switch (e.getType()) {
                                case PUT, DELETE -> {
                                    ifCacheReload = true;
                                    break g;
                                }
                            }
                        }
                    }
                }
            }
        });

    }


    protected void putKeyValue(String key, String value) {
        kvClient.put(ByteString.copyFrom(key, StandardCharsets.UTF_8), ByteString.copyFrom(value, StandardCharsets.UTF_8)).async();
    }


    protected List<KeyVal> loadKeyValByPrefix(String prefix) {
        RangeResponse result = kvClient.get(ByteString.copyFrom(prefix, StandardCharsets.UTF_8)).asPrefix().sync();
        if (result == null || result.getKvsList().isEmpty()) {
            return null;
        }
        List<KeyVal> rst = new ArrayList<>(result.getKvsList().size());
        for (KeyValue keyValue : result.getKvsList()) {
            KeyVal keyVal = new KeyVal();
            keyVal.setKey(keyValue.getKey().toStringUtf8());
            keyVal.setValue(keyValue.getValue().toStringUtf8());
            rst.add(keyVal);
        }
        return rst;
    }

    protected void deleteByKey(String key) {
        kvClient.delete(ByteString.copyFrom(key, StandardCharsets.UTF_8)).sync();
    }

    public List<String> loadValByPrefix(String prefix) {
        RangeResponse result = kvClient.get(ByteString.copyFrom(prefix, StandardCharsets.UTF_8)).asPrefix().sync();
        if (result == null || result.getKvsList().isEmpty()) {
            return null;
        }
        return result.getKvsList().stream().map(o -> {
            return o.getValue().toStringUtf8();
        }).collect(Collectors.toList());
    }


}
